#include "../common/config.h"
#include "fsconference.h"
LoggerPtr gLogger = Logger::getLogger("fsconf");
int  gRunThread = 1;


apr_queue_t* jobqueue = NULL;
apr_queue_t* resultqueue = NULL;
apr_queue_t* conn_queue = NULL;

void * APR_THREAD_FUNC fetch_order(apr_thread_t *thd, void *data);
void * APR_THREAD_FUNC result_order(apr_thread_t *thd, void *data);

void MakeDailString(vector<call_param_t>& callParams, string& strDial)
{
    string strProtocol = "sofia/external/";
    stringstream ss;
    int n = 0;

    for (vector<call_param_t>::iterator it = callParams.begin(); it != callParams.end(); ++it)
    {
        if (n > 0)
        {
            ss << "|";
        }

        ss << "[calltype=6,execute_on_answer=\'sched_hangup +" << it->nTimeout << " alloted_timeout\',origination_caller_id_name=\'" << it->caller << "\',origination_caller_id_number=" << it->caller << ",absolute_codec_string=\'" << it->codec << "\']"
        << strProtocol << it->called << "@" << it->server;
        n++;
    }

    strDial = ss.str();
}
void outcall(esl_handle_t* handle, conference_info_t* pJob)
{
    //设置拨号字符串
    stringstream ss;
    ss.str("");
    ss << "bgapi originate {ignore_early_media=true,proxy_media=true,jitterbuffer_msec=180,call_timeout=10,orig_acct=" << pJob->acct << ",orig_caller=" << pJob->caller << ",orig_called=\'" << pJob->called << "\',pin=" << pJob->passwd << "}";
    vector<call_param_t> callParams;

    if (!getCallParam(pJob->acct, pJob->caller, pJob->caller, CONFERENCE, callParams))
    {
        apr_queue_push (jobqueue, pJob);
        return;
    }

    if (callParams.begin()->nTimeout <= 0)
    {
        LOG4CXX_TRACE(gLogger, "err:timeout=" << callParams.begin()->nTimeout << "\taccount=" << pJob->acct << "\torigcaller=" << pJob->caller  << "\torigcalled=" << pJob->called );
        esl_execute(handle, "hangup", NULL, NULL);
        esl_disconnect(handle);
        return;
    }

    string dialstring;
    MakeDailString(callParams, dialstring);
    ss << dialstring << " conf XML conf " << pJob->caller << " " << pJob->caller;

    if (esl_send_recv(handle, ss.str().c_str()) != ESL_SUCCESS)
    {
        //disconnect or fs failed, retry job
        apr_queue_push (jobqueue, pJob);
        return;
    }

    pJob->result = "1";
    apr_queue_push (resultqueue, pJob);
}



static void *msg_thread_run(esl_thread_t *me, void *obj)
{

    callserver_info_t*   pCtrl = (callserver_info_t*) obj;

    esl_handle_t handle = {{0}};

    if (esl_connect(&handle, pCtrl->server, pCtrl->port, pCtrl->user, pCtrl->pwd) == ESL_SUCCESS)
    {
        LOG4CXX_INFO(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " success");
    }
    else
    {
        LOG4CXX_WARN(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " failed");
        goto done;
    }

    while (handle.connected)
    {
        conference_info_t* pJob = NULL;

        while (handle.connected && (apr_queue_pop(jobqueue, (void**)&pJob) == APR_SUCCESS))
        {
            outcall(&handle, pJob);
            pJob = NULL;
        }

        if (esl_recv_event_timed(&handle, 100, 1, NULL) == ESL_FAIL)
        {
            break;
        }
    }

done:
    apr_sleep(APR_USEC_PER_SEC);
    apr_queue_push(conn_queue, pCtrl);
    return NULL;
}





static void sighandler( int sig_no )
{
    gRunThread = 0;
}



#ifndef _DEBUG
int localMain(int argc, char* argv[]);
int main(int argc, char *argv[])
{
    if (init_daemon(argc, argv, localMain) < 0)
    {
        return 0;
    }

    return 0;
}
int localMain(int argc, char* argv[])
#else
int main(int argc, char *argv[])
#endif
{
    signal(SIGUSR1, sighandler);
    apr_initialize();
    CIniFile config;
    string strCfgPath("/etc/fsconference.ini");

    if (argc >= 2)
    {
        strCfgPath = argv[1];
    }

    ostringstream ss;

    config.SetPath(strCfgPath);
    config.ReadFile();
    PropertyConfigurator::configure(strCfgPath);

    const string password = config.GetValue("conference", "fsstr", "66call");
    const string strClient = config.GetValue("conference", "callservers", "");

    const int    nThreadPool = config.GetValueI("conference", "threads", 100);
    const int    nQueuePool = config.GetValueI("conference", "jobqueue", 1000);


    apr_thread_pool_t *thrp;


    //initalize线程池
    apr_pool_t *main_Pool;

    if (apr_pool_create(&main_Pool, NULL) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init memory pool ok");

    if (apr_queue_create(&jobqueue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_terminate();
        return -1;
    }

    if (apr_queue_create(&resultqueue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_queue_term(jobqueue);
        apr_terminate();
        return -1;
    }

    if (apr_queue_create(&conn_queue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init call job queue pool ok");

    if (apr_thread_pool_create(&thrp, nThreadPool, nThreadPool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_thread_pool_create error,thread pool=" << nThreadPool);
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_queue_term(conn_queue);
        apr_pool_destroy(main_Pool);

        apr_terminate();
        return -2;//线程池创建失败
    }

    if (!initDBS(config, "conference-db", main_Pool))
    {
        LOG4CXX_WARN(gLogger, "init dbs error");
        apr_queue_term(jobqueue);
        apr_queue_term(resultqueue);
        apr_queue_term(conn_queue);
        apr_pool_destroy(main_Pool);
        apr_terminate();
        return -3;//数据库初始化失败
    }

    //创建客户监视线程
    if (strClient.length() > 0)
    {
        string delimiter(",");
        const vector<string>& clientlist = split(strClient, delimiter);

        for (vector<string>::const_iterator iter = clientlist.begin(); iter != clientlist.end(); ++iter)
        {
            char* pIp = strdup((*iter).c_str());
            char* pPort = strchr(pIp, ':');

            if (pPort != NULL)
            {
                *pPort++ = '\0';
            }
            else
            {
                pPort = "8021";
            }

            callserver_info_t* pCtrl = new callserver_info_t();
            pCtrl->server = strdup(pIp);
            pCtrl->port = atoi(pPort);
            pCtrl->user = NULL;
            pCtrl->pwd = strdup(password.c_str());
            apr_queue_push(conn_queue, pCtrl);
            free(pIp);
        }
    }

    esl_global_set_logger(NULL);
    //创建数据库监视线程
    apr_thread_pool_push(thrp, fetch_order, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);
    apr_thread_pool_push(thrp, result_order, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);

    while (gRunThread)
    {
        callserver_info_t* pCtrl = NULL;
        apr_status_t ret = apr_queue_pop(conn_queue, (void**) & pCtrl);

        if (ret == APR_SUCCESS)
        {
            if (esl_thread_create_detached(msg_thread_run, pCtrl) != ESL_SUCCESS)
            {
                apr_queue_push(conn_queue, pCtrl);
            }
        }
        else if (ret == APR_EOF)
        {
            break;
        }
    }

    gRunThread = 0;
    apr_queue_term(jobqueue);
    apr_queue_term(resultqueue);
    apr_queue_term(conn_queue);
    apr_thread_pool_destroy(thrp);

    destoryDBS();
    apr_pool_destroy(main_Pool);
    apr_terminate();
    return 0;
}
